package com.dy.dws.HospPatGenAGConsulWin;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.dy.dws.function.MyAsyncDataStream;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsTradeHospPatGenAGConsulBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.*;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dws
 * @className: DwsTradeHospPatGenAGConsulBean
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/13 9:50
 * @version: 1.0
 * 交易域医院性别年龄段粒度问诊各窗口汇总表
 * 首先对数据进行过来 使用异步io连接维度表 医生表 和 医院表 然后进行开窗聚合 最后写入doris
 */

public class DwsTradeHospPatGenAGConsulWin extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeHospPatGenAGConsulWin().start(Constant.TOPIC_DWD_TRAFFIC_CONSULTATION, Constant.TOPIC_DWS_TRADE_HOSP_PAT_GEN_A_G_CONSUL_WIN, 4, 10019);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        //过滤出 user_id  consultation_time  doctor_id 不为空的数据 因为下游需要
        SingleOutputStreamOperator<JSONObject> eltStream = getEltStream(dataStreamSource);
//        eltStream.print();
        //转为实体类
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> beanStream = getBeanStream(eltStream);
        //连接维度表
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> hostPid = getHostPid(beanStream);
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getHospitalName = getHospotalName(hostPid);
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> patient = getPatient(getHospitalName);

        //设置水位持
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> wmStream = getWmStream(patient);
        //按照医院和年龄段进行keyby
        KeyedStream<DwsTradeHospPatGenAGConsulBean, Tuple2<String, String>> keyByStream = wmStream.keyBy(new KeySelector<DwsTradeHospPatGenAGConsulBean, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean) throws Exception {
                String hospitalId = dwsTradeHospPatGenAGConsulBean.getHospitalId();
                String ageGroup = dwsTradeHospPatGenAGConsulBean.getAgeGroup();
                return Tuple2.of(hospitalId, ageGroup);
            }
        });
//        keyByStream.print();
        //开窗聚合
        SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> result = keyByStream.window(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessWindowFunction<DwsTradeHospPatGenAGConsulBean, DwsTradeHospPatGenAGConsulBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<DwsTradeHospPatGenAGConsulBean, DwsTradeHospPatGenAGConsulBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<DwsTradeHospPatGenAGConsulBean> iterable, Collector<DwsTradeHospPatGenAGConsulBean> collector) throws Exception {
                ArrayList<DwsTradeHospPatGenAGConsulBean> dwsTradeHospPatGenAGConsulBeanArrayList = ListUtil.toList(iterable);
                double sum = dwsTradeHospPatGenAGConsulBeanArrayList.stream().mapToDouble(e -> e.getConsultationAmount().doubleValue()).sum();
                long count = dwsTradeHospPatGenAGConsulBeanArrayList.stream().mapToInt(e -> e.getConsultationCount().intValue()).sum();
                String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                String curtDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                DwsTradeHospPatGenAGConsulBean next = dwsTradeHospPatGenAGConsulBeanArrayList.iterator().next();
                DwsTradeHospPatGenAGConsulBean result = DwsTradeHospPatGenAGConsulBean.builder()
                        .hospitalId(next.getHospitalId())
                        .ageGroup(next.getAgeGroup())
                        .consultationAmount(new BigDecimal(sum))
                        .consultationCount(count)
                        .stt(start)
                        .edt(end)
                        .curDate(curtDate)
                        .userId(next.getId())
                        .createTime(next.getCreateTime())
                        .consultationFee(next.getConsultationFee())
                        .doctorId(next.getDoctorId())
                        .hospitalName(next.getHospitalName())
                        .patientId(next.getPatientId())
                        .patientGenderCode(next.getPatientGenderCode())
                        .patientGender(next.getPatientGender())
                        .id(next.getId())
                        .build();
                collector.collect(result);
            }
        });
        result.print();
//        result.map(new DorisMapFunction<>()).print();

//        result.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_hosp_pat_gen_a_g_consul_win"));


    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getWmStream(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> patient) {
        return patient.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsTradeHospPatGenAGConsulBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<DwsTradeHospPatGenAGConsulBean>() {
            @Override
            public long extractTimestamp(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean, long l) {
                return dwsTradeHospPatGenAGConsulBean.getTs();
            }
        }));
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getPatient(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getHospitalName) {
        return AsyncDataStream.unorderedWait(getHospitalName, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulBean>("dim_patient") {

            @Override
            public void getResult(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGConsulBean.setPatientGenderCode(jsonObject.getString("gender"));
                String birthday = jsonObject.getString("birthday");
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
                String s = DateFormatUtil.tsToDate(System.currentTimeMillis());
                LocalDate parse = LocalDate.parse(s, formatter);
                int year1 = parse.getYear();
                int age;
                String ageGroup = null;
                if (birthday != null) {
                    LocalDate birthDate = LocalDate.parse(birthday, formatter);
                    int year = birthDate.getYear();
                    age = year1 - year;
                    if (age >= 0 && age <= 2) {
                        ageGroup = "婴儿期";
                    } else if (age <= 5) {
                        ageGroup = "幼儿期";
                    } else if (age <= 11) {
                        ageGroup = "小学阶段";
                    } else if (age <= 17) {
                        ageGroup = "青少年期";
                    } else if (age <= 29) {
                        ageGroup = "青年期";
                    } else if (age <= 59) {
                        ageGroup = "中年期";
                    } else if (age > 60) {
                        ageGroup = "老年期";
                    } else {
                        ageGroup = "年龄异常";
                    }
                }
                dwsTradeHospPatGenAGConsulBean.setAgeGroup(ageGroup);
                dwsTradeHospPatGenAGConsulBean.setPatientGender(jsonObject.getString("gender"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean) {
                return dwsTradeHospPatGenAGConsulBean.getPatientId();
            }
        }, 5000, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getHospotalName(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> hostPid) {
        return AsyncDataStream.unorderedWait(hostPid, new MyAsyncDataStream<DwsTradeHospPatGenAGConsulBean>("dim_hospital") {
            @Override
            public void getResult(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGConsulBean.setHospitalName(jsonObject.getString("name"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean) {
                return dwsTradeHospPatGenAGConsulBean.getHospitalId();
            }
        }, 5000, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getHostPid(SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> beanStream) {
        return AsyncDataStream.unorderedWait(beanStream, new RichAsyncFunction<DwsTradeHospPatGenAGConsulBean, DwsTradeHospPatGenAGConsulBean>() {
            private AsyncConnection hBaseAsyncConnection;
            private StatefulRedisConnection<String, String> redisAsyncConnection;

            @Override
            public void open(Configuration parameters) throws Exception {
                hBaseAsyncConnection = HbaseUtil.getHBaseAsyncConnection();
                redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
            }

            @Override
            public void close() throws Exception {
                HbaseUtil.closeAsyncConnection(hBaseAsyncConnection);
                RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);
            }

            @Override
            public void asyncInvoke(DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean, ResultFuture<DwsTradeHospPatGenAGConsulBean> resultFuture) throws Exception {
                String doctorId = dwsTradeHospPatGenAGConsulBean.getDoctorId();
                String redisKey = RedisUtil.getRedisKey("dim_doctor", doctorId);
                CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        String result = null;
                        RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(redisKey);
                        try {
                            result = stringRedisFuture.get();
                        } catch (InterruptedException | ExecutionException e) {
                            resultFuture.completeExceptionally(e);
                            return null;
                        }
                        return JSONObject.parseObject(result);
                    }
                }).thenApplyAsync(new Function<JSONObject, JSONObject>() {
                    private JSONObject dimDoctor;

                    @Override
                    public JSONObject apply(JSONObject jsonObject) {
                        if (jsonObject == null) {
                            try {
                                dimDoctor = HbaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, "dim_doctor", doctorId);
                                redisAsyncConnection.async().setex(redisKey, 24 * 24 * 60, JSONObject.toJSONString(dimDoctor));
                            } catch (IOException e) {
                                resultFuture.completeExceptionally(e);
                                return null;
                            }
                        } else {
                            dimDoctor = jsonObject;
                        }
                        return dimDoctor;
                    }
                }).thenAccept(new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject jsonObject) {
                        if (jsonObject != null) {
                            dwsTradeHospPatGenAGConsulBean.setHospitalId(jsonObject.getString("hospital_id"));
                        }
                        resultFuture.complete(Collections.singleton(dwsTradeHospPatGenAGConsulBean));
                    }
                }).exceptionally(ex -> {
                    resultFuture.completeExceptionally(ex);
                    return null;
                });
            }
        }, 5000, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGConsulBean> getBeanStream(SingleOutputStreamOperator<JSONObject> eltStream) {
        return eltStream.map(new MapFunction<JSONObject, DwsTradeHospPatGenAGConsulBean>() {
            @Override
            public DwsTradeHospPatGenAGConsulBean map(JSONObject jsonObject) throws Exception {
                DwsTradeHospPatGenAGConsulBean dwsTradeHospPatGenAGConsulBean = new DwsTradeHospPatGenAGConsulBean();
                dwsTradeHospPatGenAGConsulBean.setDoctorId(jsonObject.getString("doctor_id"));
                dwsTradeHospPatGenAGConsulBean.setConsultationFee(jsonObject.getBigDecimal("consultation_fee"));
                dwsTradeHospPatGenAGConsulBean.setUserId(jsonObject.getString("user_id"));
                dwsTradeHospPatGenAGConsulBean.setPatientId(jsonObject.getString("patient_id"));
                dwsTradeHospPatGenAGConsulBean.setCreateTime(jsonObject.getString("consultation_time"));
                dwsTradeHospPatGenAGConsulBean.setId(jsonObject.getString("id"));
                dwsTradeHospPatGenAGConsulBean.setConsultationCount(1L);
                dwsTradeHospPatGenAGConsulBean.setConsultationAmount(jsonObject.getBigDecimal("consultation_fee"));
                return dwsTradeHospPatGenAGConsulBean;
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getEltStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    String user_id = jsonObject.getString("user_id");
                    String consultation_time = jsonObject.getString("consultation_time");
                    String doctor_id = jsonObject.getString("doctor_id");
                    if (user_id != null && consultation_time != null && doctor_id != null) {
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
